package com.afdor.rws.event.driven.channel.local.producer;

import com.afdor.rws.annotation.extension.Extension;
import com.afdor.rws.core.Callback;
import com.afdor.rws.event.driven.channel.local.observer.ObserverEventSource;
import com.afdor.rws.event.driven.channel.producer.AbstractEventProducer;
import com.afdor.rws.event.driven.channel.producer.EventProducer;
import com.afdor.rws.event.driven.commons.ThreadUtils;
import com.afdor.rws.event.driven.entities.Event;
import com.afdor.rws.event.driven.entities.holder.CallbackEvent;
import com.afdor.rws.event.driven.entities.holder.ResultEvent;
import lombok.Getter;
import lombok.NoArgsConstructor;
import lombok.NonNull;
import lombok.extern.slf4j.Slf4j;

import java.util.List;
import java.util.Optional;
import java.util.concurrent.CompletableFuture;

/**
 * 本地事件生产者
 *
 * @author 悭梵
 * @date Created in 2018-08-08 15:21
 */
@Slf4j
@NoArgsConstructor
@Extension(value = "local", order = Extension.NORMAL_ORDER - 100, scope = Extension.Scope.PROTOTYPE, interfaceClass = EventProducer.class)
public class LocalEventProducer extends AbstractEventProducer {

    @Getter
    private ObserverEventSource eventSource;

    public LocalEventProducer(ObserverEventSource eventSource) {
        super();
        setEventSource(eventSource);
    }

    @Override
    public void destroy() {
        this.eventSource = null;
        super.destroy();
    }

    public void setEventSource(ObserverEventSource eventSource) {
        this.eventSource = Optional.ofNullable(eventSource).orElseThrow(() -> new IllegalArgumentException("事件管理容器为空"));
    }

    @Override
    public <V> ResultEvent<V> sendEvent(Event event) {
        checkup();
        return getEventSource().sendEvent(event);
    }

    @Override
    public void sendAsyncEvent(Event event) {
        checkup();
        getEventSource().sendAsyncEvent(event);
    }

    @Override
    public <V> void asyncSendEvent(@NonNull CallbackEvent callbackEvent) {
        checkup();
        try {
            CompletableFuture<ResultEvent<V>> completableFuture = getEventSource().asyncSendEvent(callbackEvent.getEvent());
            callbackEvent.complete(completableFuture.get());
        } catch (Exception e) {
            callbackEvent.failureExceptionally(e);
        }
    }

    @Override
    public void batchSendAsyncEvent(List<Event> eventList) {
        ObserverEventSource eventSource = getEventSource();
        eventList.forEach(event -> eventSource.sendAsyncEvent(event));
    }
}
